package com.atguigu.flink.sqlfunction;

import com.atguigu.flink.function.MyUtil;
import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.UserBehavior;
import com.atguigu.flink.pojo.WaterSensor;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

/**
 * Created by Smexy on 2023/3/5
 *
 *    每隔5min输出最近1h内点击量最多的前3个商品
 *
 *    5min算一次
 *    计算的时间范围: 1h
 *
 *    slideWindow( 1h, 5min )
 *
 *    --------------------
 *      点击量最多的前3个商品
 *          ①先统计每个商品的点击量。累加求和
 *                   slideWindow( 1h, 5min )
 *               [1:00,2:00)
 *                  keyBy 商品id
 *                      a-----3
 *                      b-----4
 *                      c-----5
 *                      d-----6
 *
 *                      窗口运算，调用process，是窗口关闭时只执行一次。
 *
 *               [1:05,2:05)
 *                       a-----3
 *                       b-----2
 *
 *
 *          ②从中挑选最多的3个。 top3
 *                  keyBy 窗口id
 *                  不用。需要按照 [1:00,2:00) 分组。
 *                  没有使用窗口，process是来一条数据就运算一次！需要保证，等同一个时间段的数据全部到达，才能出发运算。
 *
 *                  在第一条数据到达时，指定一个和窗口的结束时间一致的定时器，当前到达定时器执行时间，一定可以保证上游 窗口结束时间之前
 *                  的数据都已经全部达到，可以执行运算！
 *
 *
 */
public class Demo8_TopN
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<UserBehavior> watermarkStrategy = WatermarkStrategy
            //水印会推迟3s
            .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3))
            .withTimestampAssigner((e, ts) -> e.getTimestamp() * 1000);

        //读取数据，过滤出pv数据，产生水印，按照itemId分组
        KeyedStream<UserBehavior, Long> ds1 = env
            .readTextFile("data/UserBehavior.csv")
            .map(new MapFunction<String, UserBehavior>()
            {
                @Override
                public UserBehavior map(String value) throws Exception {
                    String[] words = value.split(",");
                    return new UserBehavior(
                        Long.valueOf(words[0]),
                        Long.valueOf(words[1]),
                        Integer.valueOf(words[2]),
                        words[3],
                        Long.valueOf(words[4])
                    );
                }
            })
            .filter(u -> "pv".equals(u.getBehavior()))
            .assignTimestampsAndWatermarks(watermarkStrategy)
            //按照 商品id 分组
            .keyBy(UserBehavior::getItemId);

        //第一次聚合，开窗  slideWindow( 1h, 5min )
        SingleOutputStreamOperator<HotItem> ds2 = ds1
            .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5)))
            //滚动聚合效率高  reduce，aggregate(让输入和输出不同)
            .aggregate(new AggregateFunction<UserBehavior, Integer, HotItem>()
                       {
                           @Override
                           public Integer createAccumulator() {
                               return 0;
                           }

                           @Override
                           public Integer add(UserBehavior value, Integer accumulator) {
                               return accumulator += 1;
                           }

                           @Override
                           public HotItem getResult(Integer accumulator) {
                               return new HotItem(null, null, null, accumulator);
                           }

                           //不用实现
                           @Override
                           public Integer merge(Integer a, Integer b) {
                               return null;
                           }
                       },
                new WindowFunction<HotItem, HotItem, Long, TimeWindow>()
                {
                    //只调用一次
                    @Override
                    public void apply(Long key, TimeWindow window, Iterable<HotItem> input, Collector<HotItem> out) throws Exception {
                        HotItem result = input.iterator().next();
                        result.setStart(window.getStart());
                        result.setEnd(window.getEnd());
                        result.setItemId(key);
                        out.collect(result);
                    }
                });


        //第二次聚合
        ds2
            .keyBy(HotItem::getEnd)
            .process(new KeyedProcessFunction<Long, HotItem, String>()
            {

                private ValueState<Boolean> ifFirst;
                private ListState<HotItem> data;


                @Override
                public void open(Configuration parameters) throws Exception {
                    data = getRuntimeContext().getListState(new ListStateDescriptor<>("data", HotItem.class));
                    ifFirst = getRuntimeContext().getState(new ValueStateDescriptor<>("ifFirst", Boolean.class));
                }

                //一条数据调用一次
                @Override
                public void processElement(HotItem value, Context ctx, Collector<String> out) throws Exception {
                    TimerService timerService = ctx.timerService();
                    //当前是这个 窗口的第一条数据到达
                    if (ifFirst.value() == null){
                        timerService.registerEventTimeTimer(value.getEnd() + 5000);
                        ifFirst.update(false);
                    }

                    //放入集合，暂存
                    data.add(value);

                }

                @Override
                public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

                    List<HotItem> top3 = StreamSupport.stream(data.get().spliterator(), true)
                                                         .sorted((h1, h2) -> -h1.getClick().compareTo(h2.getClick()))
                                                         .limit(3)
                                                         .collect(Collectors.toList());

                    HotItem hotItem = top3.get(0);

                    List<Integer> top3Click = top3.stream().map(h -> h.getClick()).collect(Collectors.toList());

                    out.collect(hotItem.getItemId() + ":" + MyUtil.parseTimeWindow(new TimeWindow(hotItem.getStart(),hotItem.getEnd()))
                            +"  " + top3Click);

                }
            })
            .print();


        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }

    @AllArgsConstructor
    @NoArgsConstructor
    @Data
    public static class HotItem{

        private Long start;
        private Long end;
        private Long itemId;
        private Integer click;

    }
}
